use anyhow::Result;
use chrono;
use chrono::{Datelike, Timelike};
use sqlx::{types, MySql, Pool};
use std::ops::Sub;
use crate::miner::types::TaskType;

#[derive(sqlx::FromRow, Debug, Clone)]
pub struct SealTaskDetail {
    pub task_id: String,
    pub task_type: String,
    pub sector_number: i64,
    pub worker_host: String,
    pub assign_time: chrono::NaiveDateTime,
    pub receive_time: Option<chrono::NaiveDateTime>,
    pub complete_time: Option<chrono::NaiveDateTime>,
    pub task_status: i32,
}

#[derive(sqlx::FromRow, Debug)]
pub struct Worker {
    pub worker_id: String,
    pub worker_url: String,
    pub worker_host: String,
    pub worker_enable: i32,
    pub accept_new: i32,
    pub update_time: chrono::NaiveDateTime,
}

#[derive(sqlx::FromRow, Debug)]
pub struct Sector {
    pub sector_number: i64,
    pub miner_actor: i64,
    pub sector_stage: String,
    pub worker_host: String,
    pub update_time: chrono::NaiveDateTime,
    pub sector_status: i32,
}

#[derive(sqlx::FromRow, Debug)]
pub struct CommonCount {
    pub data_count: i64,
}

#[derive(sqlx::FromRow, Debug)]
pub struct CommonString {
    pub content: String,
}

#[derive(sqlx::FromRow, Debug)]
pub struct CommonWorkerCount {
    pub worker_id: String,
    pub task_count: i32,
}

#[derive(sqlx::FromRow, Debug)]
pub struct CommonWorkerLimit {
    pub worker_id: String,
    pub task_limit: i32,
}

#[derive(sqlx::FromRow, Debug, Clone)]
pub struct WorkerJobs {
    pub task_id: String,
    pub task_type: String,
    pub sector_number: i64,
    pub worker_id: String,
    pub worker_host: String,
    pub assign_time: chrono::NaiveDateTime,
    pub receive_time: Option<chrono::NaiveDateTime>,
    pub task_status: i32,
}

pub async fn query_seal_task(pool: &Pool<MySql>) -> Result<Vec<SealTaskDetail>> {
    let now = chrono::Local::now();
    let now = now.sub(chrono::Duration::hours(2));
    let assign = types::chrono::NaiveDate::from_ymd(now.year(), now.month(), now.day()).and_hms(
        now.hour(),
        now.minute(),
        now.second(),
    );
    // let mut works = Vec::<SealTaskDetail>::new();
    let query = "select task_id, task_type, sector_number, worker_host, assign_time, receive_time, complete_time, task_status from seal_task where assign_time < ? and task_status < 2";

    let array: Vec<SealTaskDetail> = sqlx::query_as::<MySql, SealTaskDetail>(query)
        .bind(assign)
        .fetch_all(pool)
        .await?;
    Ok(array)
}

pub async fn query_db_running_task(pool: &Pool<MySql>) -> Result<Vec<SealTaskDetail>> {
    let now = chrono::Local::now();
    let now = now.sub(chrono::Duration::minutes(10));
    let running = types::chrono::NaiveDate::from_ymd(now.year(), now.month(), now.day()).and_hms(
        now.hour(),
        now.minute(),
        now.second(),
    );
    let query = "select task_id, task_type, sector_number, worker_host, assign_time, receive_time, complete_time, task_status from seal_task where assign_time < ? and task_status < 2";

    let array: Vec<SealTaskDetail> = sqlx::query_as::<MySql, SealTaskDetail>(query)
        .bind(running)
        .fetch_all(pool)
        .await?;
    Ok(array)
}

pub async fn set_db_task_fail(pool: &Pool<MySql>, job: &SealTaskDetail) -> Result<u64> {
    let query = "update seal_task set complete_time = sysdate(), task_status = 3 where sector_number = ? and task_type = ? and task_status < 2";

    let ret = sqlx::query(query)
        .bind(&job.sector_number)
        .bind(&job.task_type)
        .execute(pool)
        .await?;
    Ok(ret.rows_affected())
}

pub async fn pledge_sector_for_worker(
    pool: &Pool<MySql>,
    worker: &String,
    sector: u64,
    miner: u64,
) -> Result<u64> {
    let sql = "insert into sector (sector_number, miner_actor, sector_stage, worker_host, update_time, sector_status) values (?, ?, 'NEW', ?, sysdate(), 1)";
    let ret = sqlx::query(sql)
        .bind(sector)
        .bind(miner)
        .bind(worker)
        .execute(pool)
        .await?;
    Ok(ret.rows_affected())
}

pub async fn is_sector_assigned_to_worker(
    pool: &Pool<MySql>,
    sector: u64,
    miner: u64,
) -> Result<Option<CommonString>> {
    let sql = "select worker_id as content from sector where sector_number = ? and miner_actor = ?";
    let worker: Option<CommonString> = sqlx::query_as::<_, CommonString>(sql)
        .bind(sector)
        .bind(miner)
        .fetch_optional(pool)
        .await?;
    Ok(worker)
}

pub async fn query_worker_pc1_sector_count(pool: &Pool<MySql>, worker: &str) -> Result<u64> {
    let sql = "select count(1) as data_count from sector where worker_host = ? and sector_stage in ('NEW', 'F', 'AP', 'PC1') and sector_status = 1";
    let count: CommonCount = sqlx::query_as::<_, CommonCount>(sql)
        .bind(worker)
        .fetch_one(pool)
        .await?;
    Ok(count.data_count as u64)
}

pub async fn query_worker_info_with_id(pool: &Pool<MySql>, worker: &str) -> Result<Option<Worker>> {
    let sql = "select worker_id, worker_url, worker_host, worker_enable, accept_new, update_time from worker where worker_id = ?";
    let w: Option<Worker> = sqlx::query_as::<_, Worker>(sql)
        .bind(worker)
        .fetch_optional(pool)
        .await?;
    Ok(w)
}

pub async fn repair_wrong_p1_sector(pool: &Pool<MySql>) -> Result<()> {
    let query = "select * from sector where sector_stage = 'PC1' and sector_number between 26000 and 30000";
    let array: Vec<Sector> = sqlx::query_as::<MySql, Sector>(query)
        .fetch_all(pool)
        .await?;
    let update = "insert into seal_task (task_id, task_type, sector_number, miner_actor, worker_id, assign_time, receive_time, complete_time, task_status)
values (uuid(), 'AP', ?, 808701, '3f6e3cdf-8974-429a-b2e2-92e121270f5f', sysdate(), sysdate(), sysdate(), 2)";
    for sector in array {
        let number = sector.sector_number;
        // let _ = sqlx::query(update).bind(number).execute(pool).await?;
        log::info!("ln -s /opt/local_ssd/addpiece_cache_64g/sector_cache.dat /opt/hdd_pool/.lotusworker/unsealed/s-t0808701-{}", number);
    }
    Ok(())
}

pub async fn update_worker_id_of_sector_table(pool: &Pool<MySql>, worker: &str, sector: u64, ty: TaskType) -> Result<u64> {
    let sql = "update sector_copy1 set worker_id = ? where sector_number = ? and sector_stage = ? and worker_id = '6c1eb101-dfad-4fff-85c6-136c6578b714' and update_time < '2021-10-07 19:00:00'";
    let ret = sqlx::query(sql)
        .bind(worker)
        .bind(sector)
        .bind(ty.short())
        .execute(pool)
        .await?;
    Ok(ret.rows_affected())
}


pub async fn query_p1_sector_list(pool: &Pool<MySql>) -> Result<Vec<Sector>> {
    let sql = "select sector_number, miner_actor, sector_stage, worker_id, update_time, sector_status from sector where sector_stage = 'PC1'";
    let w: Vec<Sector> = sqlx::query_as::<_, Sector>(sql)
        .fetch_all(pool)
        .await?;
    Ok(w)
}

pub async fn update_sector_status(pool: &Pool<MySql>, sector: u64) -> Result<u64> {
    let sql = "update sector set sector_stage = 'remove' where sector_number = ?";
    let ret = sqlx::query(sql)
        .bind(sector)
        .execute(pool)
        .await?;
    Ok(ret.rows_affected())
}

pub async fn query_worker_seal_task_count(pool: &Pool<MySql>) -> Result<Vec<CommonWorkerCount>>{
    let sql = "select worker_id, count(1) as task_count from seal_task where task_type = 'PC1' and task_status < 2 group by worker_id";
    let w: Vec<CommonWorkerCount> = sqlx::query_as::<_, CommonWorkerCount>(sql)
        .fetch_all(pool)
        .await?;
    Ok(w)
}

pub async fn query_worker_sector_count(pool: &Pool<MySql>) -> Result<Vec<CommonWorkerCount>>{
    let sql = "select worker_id, count(1) as task_count from sector where sector_stage = 'PC1' group by worker_id";
    let w: Vec<CommonWorkerCount> = sqlx::query_as::<_, CommonWorkerCount>(sql)
        .fetch_all(pool)
        .await?;
    Ok(w)
}

pub async fn query_worker_task_limit(pool: &Pool<MySql>) -> Result<Vec<CommonWorkerLimit>>{
    let sql = "select w.worker_id, c.task_limit from worker w, worker_config c where w.worker_host=c.worker_host and c.task_type = 'PC1'";
    let w: Vec<CommonWorkerLimit> = sqlx::query_as::<_, CommonWorkerLimit>(sql)
        .fetch_all(pool)
        .await?;
    Ok(w)
}

pub async fn query_sector_running_in_db(pool: &Pool<MySql>) -> Result<Vec<WorkerJobs>> {
    let query = "select w.worker_id, t.worker_host, t.task_id, task_type, sector_number, miner_actor, assign_time, receive_time, task_status from worker w, seal_task t where w.worker_host = t.worker_host and t.task_status = 1";

    let array: Vec<WorkerJobs> = sqlx::query_as::<MySql, WorkerJobs>(query)
        .fetch_all(pool)
        .await?;
    Ok(array)
}

pub async fn query_worker_sector_running(pool: &Pool<MySql>, worker: &str) -> Result<Vec<WorkerJobs>> {
    let query = "select w.worker_id, t.worker_host, t.task_id, task_type, sector_number, miner_actor, assign_time, receive_time, task_status from worker w, seal_task t where w.worker_host = t.worker_host and t.task_status = 1 and (w.worker_id like ? or w.worker_host like ?)";
    let like = format!("%{}%", worker);
    let array: Vec<WorkerJobs> = sqlx::query_as::<MySql, WorkerJobs>(query)
        .bind(&like)
        .bind(&like)
        .fetch_all(pool)
        .await?;
    Ok(array)
}

pub async fn query_workers_list(pool: &Pool<MySql>) -> Result<Vec<Worker>> {
    let query = "select worker_id, worker_url, worker_host, worker_enable, accept_new, update_time from worker";

    let array: Vec<Worker> = sqlx::query_as::<MySql, Worker>(query)
        .fetch_all(pool)
        .await?;
    Ok(array)
}

pub async fn query_running_sectors(pool: &Pool<MySql>, host: Option<&str>) -> Result<Vec<Sector>> {
    let array = if host.is_none(){
        let query = "select sector_number, miner_actor, sector_stage, sector_status, worker_host, update_time, sector_status from sector where sector_stage not in ('NEW', 'GET', 'remove') and sector_status = 1";
        let array: Vec<Sector> = sqlx::query_as::<MySql, Sector>(query)
            .fetch_all(pool)
            .await?;
        array
    } else {
        let query = "select sector_number, miner_actor, sector_stage, sector_status, worker_host, update_time, sector_status from sector where sector_stage not in ('NEW', 'GET', 'remove') and sector_status = 1 and worker_host = ?";
        let host = host.unwrap();
        let array: Vec<Sector> = sqlx::query_as::<MySql, Sector>(query)
            .bind(host)
            .fetch_all(pool)
            .await?;
        array
    };


    Ok(array)
}

pub async fn disable_running_sector(pool: &Pool<MySql>, sector: i64) -> Result<u64> {
    let query = "update sector set sector_status = 0 where sector_number = ? and sector_stage not in ('NEW', 'GET', 'remove')";

    let ret = sqlx::query(query)
        .bind(sector)
        .execute(pool)
        .await?;
    Ok(ret.rows_affected())
}

pub async fn complete_running_sector_job(pool: &Pool<MySql>, sector: i64) -> Result<u64> {
    let query = "update seal_task set complete_time = sysdate(), task_status = 3 where sector_number = ? and task_status = 1";

    let ret = sqlx::query(query)
        .bind(sector)
        .execute(pool)
        .await?;
    Ok(ret.rows_affected())
}